package anli

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**程序优化改进方案1：
 *   1、actionRDD重复使用，直接把重复使用的东西放在Driver中，即actionRDD.cache()
 *
 *   2、cogroup过程中存在shuffle操作，有可能降低性能,可将cogroup改为union并集
 * 程序优化改进方案2（在1的基础上改进）：
 *   1、将数据转换结构： 点击的场合结构 (品类ID,(1,0,0)),
 *                    下单场合结构 (品类ID ,(0,1,0))
 *                    支付场合结构 (品类ID ,(0,1,0))
 *    2、将相同的品类ID的数据进行分组聚合：(品类ID,(点击数量，下单数量，支付数量))
 *    3、将统计结果根据降序进行排序，取前十
 *   */
object HotCategoryTop10Analysis_Gai1 {
  /**
   * 网站用户行为：搜索，点击，下单，支付
   * 数据字段：日期 用户id SessionID 页面ID 动作时间 搜索关键字
   * 需求:1、读取原始日志数据
   * 2、统计品类的点击数量：品类ID和点击次数
   * 3、统计品类的下单数量：品类ID，下单数量
   * 4、统计品类的支付数量：品类ID，支付数量
   * 5、将品类进行排序，并取前10名
   *    点击数量排序，下单数量排序，支付数量排序
   *    元组排序：先比较第一个，再比较第二个，再比较后面的，依此类推
   * 6、将结果采集到控制台打印出来
   * */
  def main(args: Array[String]): Unit = {
    val sparkconf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("anli2")
    val sc=new SparkContext(sparkconf)
    // 需求:1、读取原始日志数据
    val actionRDD=sc.textFile("data/user_visit_action.txt")
    //优化：直接把重复使用的东西放在Driver中
//    actionRDD.cache()
//    1、将数据转换结构： 点击的场合结构 (品类ID,(1,0,0)),
//    *                    下单场合结构 (品类ID ,(0,1,0))
//    *                    支付场合结构 (品类ID ,(0,1,0))

    val flatRDD: RDD[(String, (Int, Int, Int))] = actionRDD.flatMap(
      action => {
        val datas = action.split("_")
        if (datas(6) != "-1") {
          //点击场合
          List((datas(6), (1, 0, 0)))
        } else if (datas(8) != "null") {
          //下单场合
          val ids = datas(8).split(",")
          ids.map(id => (id, (0, 1, 0)))
        } else if (datas(10) != "null") {
          //支付场景
          val ids = datas(10).split(",")
          ids.map(id => (id, (0, 0, 1)))
        } else {
          Nil
        }
      }
    )
    println("点击的场合结构 (品类ID,(1,0,0)),\n 下单场合结构 (品类ID ,(0,1,0))\n 支付场合结构 (品类ID ,(0,1,0))")
    flatRDD.collect().foreach(println)

//    *    2、将相同的品类ID的数据进行分组聚合：(品类ID,(点击数量，下单数量，支付数量))
val analysisRDD: RDD[(String, (Int, Int, Int))] = flatRDD.reduceByKey(
  (t1, t2) => {
    (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
  }
)
    println("将数据按照 (品类ID)，（点击数 下单数 支付数量）统计数据输出,默认按下单数据排序")
    analysisRDD.collect().foreach(println)


//    *    3、将统计结果根据降序进行排序，取前十





    //6、将结果采集到控制台打印出来





    sc.stop()
  }

}
